package com.allen.flink.batch;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple3;

/**
 * 功能: 读取Csv示例
 * types 配置字段的类型,对于已解析字段必须配置
 * lineDelimiter 指定行分隔符 默认\n
 * fieldDelimiter 指定字段分隔符 默认,
 * includeFields 指定要去读取的字段 默认由types来决定
 * parseQuotedStrings 启用引用字符串解析 默认不启用
 * ignoreComments 指定注释前缀. 忽略注释
 * ignoreInvalidLines 忽略无法正常解析的行 默认不忽略直接抛异常
 * ignoreFirstLine 忽略首行 默认不忽略
 *
 * @date: 2020-03-24 12:21
 * @author: Allen
 * @version: 0.0.4-snapshot
 * @Email: allenZyhang@163.com
 * @since: JDK 1.8
 **/
public class ReadCsv {
    
    public static void main(String[] args) throws Exception {
        //获取执行环境
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        
        /**
         * uid,age,sex,career,city
         * ##用户id,年龄,性别,职业,城市
         * 101,28,male,IT,beijing
         * 104,null,male,teacher,shanghai
         * 102,32,female,lawyer,hebei
         * 103,18,female,student,tianjing
         */
        
        //读取输入数据
        DataSet<Tuple3<Integer, Integer, String>> csvDs = env.readCsvFile("D:/Flink/user.csv").includeFields("11100").ignoreFirstLine().ignoreInvalidLines().ignoreComments("##")
                                                             .lineDelimiter("\n").fieldDelimiter(",").types(Integer.class, Integer.class, String.class);
       
        csvDs.print();
    }
    
}
